package net.bwie.realtime.jtp.dws.douyin.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.dws.douyin.log.bean.TradeMetric;
import net.bwie.realtime.jtp.dws.douyin.log.bean.TradePayInfo;
import net.bwie.realtime.jtp.dws.douyin.log.functions.DouYinWindowFunction;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import javax.naming.Context;
import java.util.HashSet;
import java.util.Set;

public class DataDwdEvent {
    public static void main(String[] args) throws Exception{
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);
        //2.数据源
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "dwd_trade_pay_detail");
//        kafkaDataStream.print();
        //3.数据转换
        DataStream<String> resultStream = handle(kafkaDataStream);
        resultStream.print();
        //4.数据输出
        DorisUtil.saveToDoris(
                resultStream,"douyin_realtime_report","dws_trade_pay_detail"
        );
        //5.触发执行
        env.execute("DataDwdEvent");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        SingleOutputStreamOperator<TradePayInfo> mapStream = stream.map(new MapFunction<String, TradePayInfo>() {
            @Override
            public TradePayInfo map(String value) throws Exception {
                return JSON.parseObject(value, TradePayInfo.class);
            }
        });

        // 过滤支付成功的记录
        SingleOutputStreamOperator<TradePayInfo> filteredStream = mapStream.filter(pay -> "success".equals(pay.getPayStatus()));

        // 窗口 滑动窗口，窗口长度是为5分钟，滑动步长为1分钟
        WindowedStream<TradePayInfo, Long, TimeWindow> windowStream = filteredStream.keyBy(TradePayInfo::getAnchorId)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)));

        //聚合计算
        SingleOutputStreamOperator<TradeMetric> applyStream = windowStream.apply(new DouYinWindowFunction());

        SingleOutputStreamOperator<String> resultStream = applyStream.map(tradeMetric -> {
            return String.format(
                    "{\"anchor_id\":\"%d\",\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"order_count\":\"%d\",\"user_count\":\"%d\"}",
                    tradeMetric.getAnchorId(),
                    tradeMetric.getWindowStartTime(),
                    tradeMetric.getWindowEndTime(),
                    tradeMetric.getOrderCount(),
                    tradeMetric.getUserCount()
            );
        });

        return resultStream;
    }
}
